package com.hex.ds.hdtp.core.app.data.transfer.handler.impl;

import cn.hutool.core.util.StrUtil;
import com.hex.ds.hdtp.core.app.common.util.JaveShellUtil;
import com.hex.ds.hdtp.core.app.data.configuration.service.ITableConvertService;
import com.hex.ds.hdtp.core.app.data.contrast.service.IContrastTableStructureService;
import com.hex.ds.hdtp.core.app.data.transfer.converter.IDataTransferLogConverter;
import com.hex.ds.hdtp.core.app.data.transfer.converter.IDataTransferSqlStructureConverter;
import com.hex.ds.hdtp.core.app.data.transfer.dto.DataTransferSqlStructureDto;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferListService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferLogService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferTaskService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceInfoService;
import com.hex.ds.hdtp.core.inf.data.contrast.service.impl.DataInfoSelectHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

@Slf4j
@Service("HIVE_HIVE_FLINK")
public class HiveHiveTransferHandlerFlinkService extends AbstractDataTransferHandlerFlinkService {

    public HiveHiveTransferHandlerFlinkService(IDataTransferTaskService dataTransferTaskService, IDataTransferListService dataTransferListService, IDataTransferLogService dataTransferLogService, IDataTransferLogConverter dataTransferLogConverter, ITableConvertService tableConvertService, IDataTransferSqlStructureConverter dataTransferSqlStructureConverter, IContrastTableStructureService contrastTableStructureService, DataInfoSelectHandlerContext dataInfoSelectHandlerContext, JaveShellUtil javaShellUtil, ISourceInfoService sourceInfoService) {
        super(dataTransferTaskService, dataTransferListService, dataTransferLogService, dataTransferLogConverter, tableConvertService, dataTransferSqlStructureConverter, contrastTableStructureService, dataInfoSelectHandlerContext, javaShellUtil, sourceInfoService);
    }

    @Override
    protected String buildCreateSourceTable(DataTransferSqlStructureDto dataTransferSqlStructureDto) {
        StringBuilder hiveSourceCatalog = new StringBuilder();
        hiveSourceCatalog.append(LINEFEED).append("CREATE CATALOG SOURCE_HIVE WITH (").append(LINEFEED)
                .append("  'type' = 'hive',").append(LINEFEED)
                .append(StrUtil.format("  'default-database' = '{}',", dataTransferSqlStructureDto.getSourceDatabase())).append(LINEFEED)
                .append(StrUtil.format("  'hive-conf-dir' = '{}'", dataTransferSqlStructureDto.getSourceConfList().get(5).getConfValue())).append(LINEFEED)
                .append(");").append(LINEFEED);
        return hiveSourceCatalog.toString();
    }

    @Override
    protected String buildCreateTargetTable(DataTransferSqlStructureDto dataTransferSqlStructureDto) {
        StringBuilder hiveTargetCatalog = new StringBuilder();
        hiveTargetCatalog.append(LINEFEED).append("CREATE CATALOG TARGET_HIVE WITH (").append(LINEFEED)
                .append("  'type' = 'hive',").append(LINEFEED)
                .append(StrUtil.format("  'default-database' = '{}',", dataTransferSqlStructureDto.getTargetDatabase())).append(LINEFEED)
                .append(StrUtil.format("  'hive-conf-dir' = '{}'", dataTransferSqlStructureDto.getTargetConfList().get(5).getConfValue())).append(LINEFEED)
                .append(");").append(LINEFEED)
                .append("USE CATALOG TARGET_HIVE;").append(LINEFEED)
                .append("-- 切换到hive方言").append(LINEFEED)
                .append("SET 'table.sql-dialect' = 'hive';").append(LINEFEED).append(LINEFEED);
        return hiveTargetCatalog.toString();
    }

}
